package com.burges.net.dataStream.codeRuler.dataInputStream.datasources

import org.apache.flink.api.java.io.CsvInputFormat
import org.apache.flink.core.fs.Path
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

/**
  * 文件作为数据源
  * 创建人    BurgessLee 
  * 创建时间   2020/1/25 
  * 描述
  */
object FileDataSourceDemo {

	def main(args: Array[String]): Unit = {
		val env = StreamExecutionEnvironment.getExecutionEnvironment

		//1.readTextFile方法
		val textStream = env.readTextFile("/usr/local/data_example.log")


		//2.readFile参数 文件读取类型WatchType 检测文件变换时间间隔interval 文件路径过滤条件FilePathFilter等
		/**
		  * 文件读取类型两种模式：
		  *    PROCESS_CONTINUOUSLY:
		  *         一旦文件内容发生变化，会将全部内容加载到Flink中处理
		  *         无法实现Excatly Once级别的数据一致性保障
		  *    PROCESS_ONCE:
		  *         文件发生变化的时候，会将变化的数据读取到Flink中，智慧呗读取和处理一次
		  *         可以保证数据Exactly Once级别的一致性保证
		  */
		/**
		  * 如果是文件作为数据源，一旦摸个节点出现故障，但是数据继续产生，CheckPoints不会更新，形成数据积压，导致恢复的时间也变长
		  */
		env.readFile(new CsvInputFormat[String](new Path("/user/local/data_example.csv")) {
			override
			def fillRecord(out: String, Objects: Array[AnyRef]): String = {
				return null
			}
		}, "/user/local/data_examples.csv")
	}

}
